package com.bigdata.hpsk.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  *  Spark Application 编程模板
  */
object SrcTrackLogAnalyzerSpark {

  /**
    * 如果Spark Application运行在本地的话，Driver Program
    *     JVM Process
    */
  def main(args: Array[String]): Unit = {

    /**
      * 从前面spark-shell命令行可知：
      *   Spark 数据分析的程序入口SparkContext，用于读取数据
      */
    // 读取Spark Application的配置信息
    val sparkConf = new SparkConf()
        // 设置SparkApplication名称
        .setAppName("TrackLog Analyzer Application")
        // 设置程序运行的环境，通常情况下，在IDE中开发的时候，设置为local mode，至少是两个Thread
        // 在实际部署的时候通过通过提交应用的命令悍进行设置
        .setMaster("local[2]")
        // 设置记录此应用的EventLog
        .set("spark.eventLog.enabled", "true")
        .set("spark.eventLog.dir", "hdfs://bigdata-training01.hpsk.com:8020/datas/spark/eventLogs/")
        .set("spark.eventLog.compress", "true")
    // 创建SparkContext上下文对象
    val sc = SparkContext.getOrCreate(sparkConf)

    // 设置日志级别
    sc.setLogLevel("WARN")

/** =================================================================== */
    /**
      * 读取要处理的数据，从HDFS上读取, 可知数据文件中每行数据中各个字段使用 tab 分割
      */
    val trackRDD: RDD[String] = sc.textFile("/user/hive/warehouse/db_track.db/yhd_log/date=20150828", 4)
    // 测试
    println(s"Count = ${trackRDD.count()} \n ${trackRDD.first()}")


    /**
      *  分析数据，需求如下：
      *     统计每日的PV和UV
      *       PV：页面访问量/浏览量
      *         pv = COUNT(url)  url 不能为空  url.length > 0  第2列
      *       UV：访客数
      *         uv = COUNT(DISTINCT guid)                     第6列
      *       时间：
      *         用户访问网站的时间字段
      *           tracktime -> 2015-08-28 18:10:00            第18列
      */
    // 清洗数据，提取字段
    val filteredRDD: RDD[(String, String, String)] = trackRDD
      // 防止空字符串，每一行
      .filter(line => line.trim.length > 0)
      // 分割数据，提取字段
      .map(line => {
        // 分割数据
        val splited = line.split("\t")
        // return -> (date, url, guid)  三元组
        (splited(17).substring(0, 10), splited(1), splited(5))
      })

    // RDD数据放到内存中，后面对RDD进行重复使用, Executor内存比较大情况下
    // filteredRDD.cache()
    filteredRDD.persist(StorageLevel.MEMORY_AND_DISK)

    /**
      * 统计每日的PV
      */
    val pvRDD = filteredRDD
      // 提取字段, 此处采用编写代码风格为case
      .map{
        case(date, url, guid) => (date, url)
      }
      // 判断URL不能为空
      .filter(tuple => tuple._2.trim.length > 0)
      // 只要 date 有访问记录，就表示用户浏览一次网站
      .map{
        case (date, url) => (date, 1)
      }
      // 此时转换为WordCount词频统计，按照日期进行聚合统计
      .reduceByKey(_ + _)

    // 打印结果RDD中的数据
    // 如果程序运行在Cluster上的时候，下面的打印信息显示在Executor的标出输出日志中。
    pvRDD.foreach(println)
    // (2015-08-28,69197)


    /**
      * 统计每日UV
      */
    val uvRDD = filteredRDD
      // 提取字段, 此处采用编写代码风格为case
      .map{
        case(date, url, guid) => (date, guid)
      }
      // 去重，相同的uig在某一天如果访问多次的话，仅仅算一次
      .distinct()
      // 只要 date 有访问记录，就表示用户当天浏览网站
      .map{
        case (date, guid) => (date, 1)
      }
      // 此时转换为WordCount词频统计，按照日期进行聚合统计
      .reduceByKey(_ + _)

    // 打印结果RDD中的数据
    uvRDD.foreach(println)
    // (2015-08-28,39007)

    // 应该讲缓存中的进行释放
    filteredRDD.unpersist()


    /**
      * 熟悉一下常用RDD的API
      */
    // def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
    val joinRDD: RDD[(String, (Int, Int))] = pvRDD.join(uvRDD)
    joinRDD.foreach{
      case (track_date, (pv, uv)) => println(s"${track_date} \t ${pv} \t ${uv}")
    }

    // def union(other: RDD[T]): RDD[T]
    val unionRDD: RDD[(String, Int)] = pvRDD.union(uvRDD)
    unionRDD.foreach(println)

    /**
      * unionRDD 结果保存到外部存储系统中，比如hdfs或者rdbms数据库表中
      *     假设unionRDD中有14 条记录，2各分区的数据
      *     - foreach： 需要创建14 数据库连接
      *     - foreachPartition： 需要创建2 数据库连接
      */
    println(s"pvRDD 分区数 = ${pvRDD.partitions.size}, uvRDD 分区数 = ${uvRDD.partitions.size}")
    println(s"UnionRDD 分区数 = ${unionRDD.partitions.size}")

    unionRDD.coalesce(1).foreach(item => {
      // 打印元素的值
      println(item)
      // 比如讲分析的结果输出到RDBMS表中（比如MySQL表中）
      // TODO: 1. 创建连接Connection
      // TODO: 2. 将每条数据插入到数据库表中
      // TODO: 3. 关闭数据库连接
    })

    println("-----------------------------------")
    unionRDD.coalesce(1).foreachPartition(iter => {
      // TODO: 1. 创建连接Connection
      // 对每个分区数据进行处理
      iter.foreach(item => {
        // 显示元素
        println(item)
        // TODO: 2. 将每条数据插入到数据库表中
      })
      // TODO: 3. 关闭数据库连接
    })

    // 判断RDD是否为空
    println(unionRDD.isEmpty())

    /** =================================================================== */
    // 在开发测试的时候，为了在每个Application页面监控查看应用中Job的运行
    Thread.sleep(10000000)

    // 关闭资源
    sc.stop()
  }

}
